import json
from kafka import KafkaProducer
from common.utils.tasks import AsyncTask
from common.conf import LOG_TOPIC_SAVE_LOG, LOG_TOPIC_SAVE_USER_ACTION, KAFKA_BOOTSTRAP_SERVERS

async_task = None


def init_async_task():
    global async_task
    async_task = AsyncTask()


def get_async_task():
    return async_task


web_producer = KafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS)


def save_log(content, level='info'):
    content = {
        'content': content,
        'level': level,
    }
    content = json.dumps(content, ensure_ascii=False).encode()

    future = web_producer.send(LOG_TOPIC_SAVE_LOG, content)
    future.get(timeout=120)


def save_user_action(request, resp):
    url = request.request.path
    method = request.request.method
    req = {
        'url': url,
        'method': method,
        'param': request.request.arguments,
    }

    content = {
        'req': req,
        'resp': resp,
        'user': json.dumps({
            'user_id': request.session.get('user_id', ''),
            'user_name': request.session.get('user_name', ''),
            'user_phone': request.session.get('user_phone', ''),
            'role': request.session.get('role', ''),
        })
    }
    content = json.dumps(content, ensure_ascii=False).encode()

    future = web_producer.send(LOG_TOPIC_SAVE_USER_ACTION, content)
    future.get(timeout=120)
